import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.Socket;
import java.net.SocketException;
import java.util.Random;
import java.util.Vector;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


import com.sleepycat.je.DatabaseException;
import com.sun.corba.se.spi.orbutil.threadpool.ThreadPool;


public class Slave extends Machine implements ISlave,ISlaveConnection {

	BerkeleyDBManager dbManager = null;
	ConnectionManager connManager = null;
	Vector<ClientHandler> cHandlers = null;
	public Thread listenerThread = null;
	int clientListenerPort = PORTS.SLAVE_LISTENER_PORT_FOR_CLIENT;
	ExecutorService execService = null;
	
	int keyStart = 0,keyEnd = 0;//this slave is responsible for keys >= keystart and <= keyend
	
	public Slave(String ip,String id,String masterip) {
		// TODO Auto-generated constructor stub
		setIp(ip);
		setId(id);
		cHandlers = new Vector<ClientHandler>();
		initialize(ip,id,masterip);
	}
	
	@Override
	public void handleMessage(IMessage message) {
		//System.out.println("Got Message : " + message);
		/*if(message.getMessageID() == IMessage.PUTDATA){
			//store locally the data
			
			//then send the response to client, so that it can cache
			IMessage msg = new Message();
			msg.setMessageID(IMessage.PUTDATA_RESPONSE);
			msg.setDest(message.getSource());
			msg.setSource(this);
			msg.setKey(new DataObject(this.getID().getBytes()));
			msg.setValue(new DataObject(this.getIP().getBytes()));
			this.sendData(msg, msg.getDest());
			
		}
		else if(message.getMessageID() == IMessage.GETDATA){
			//retrieve the data
			DataObject value = new DataObject("world".getBytes());
			//send the response to client.
			IMessage msg = new Message();
			msg.setMessageID(IMessage.GETDATA_RESPONSE);
			msg.setDest(message.getSource());
			msg.setSource(this);
			msg.setKey(message.getKey());
			msg.setValue(value);
			this.sendData(msg, msg.getDest());
		}*/
		if(message.getMessageID() == IMessage.SLAVE_JOIN_ACKNOWLEDGE){
			keyStart = Message.getIntForBytes(message.getKey().getVal());
			keyEnd = Message.getIntForBytes(message.getValue().getVal());
			if((keyStart - keyEnd) == 1)
				keyEnd = keyStart;
			//currently do nothing
		}
		else if(message.getMessageID() == IMessage.KEYSPACE_SPLIT){
			//have to do data transfer to new slave and reset keystart and keyend
			String newSlaveIP = new String(message.getKey().getVal());
			String valueString = new String(message.getValue().getVal());
			String[] values = valueString.split(":");
			
			keyStart = new Integer(values[0]).intValue();
			keyEnd = new Integer(values[1]).intValue();
			if((keyStart - keyEnd) == 1)
				keyEnd = keyStart;
			String newSlavePort = values[2];
			
			//should do the data transfer now
		}

	}
	
	public boolean validate(int key)
	{
		//System.out.println("key : " + key + " start: " + keyStart + " end: " + keyEnd);
		if((keyStart == keyEnd))//responsible for the entire keyspace
			return true;
		if(keyStart < keyEnd && key >= keyStart && key <= keyEnd)
			return true;
		if(keyStart > keyEnd){
			if(key > keyEnd && key < keyStart)
				return false;
			return true;
		}		
		return false;
	}

	public void initialize(String ip,String id,String masterip) {
				
		Random r = new Random();
		clientListenerPort = PORTS.SLAVE_LISTENER_PORT_FOR_CLIENT + r.nextInt(1000);
		connManager = new ServerConnectionManager(clientListenerPort);
		listenerThread = new SlaveListener(this,ip,masterip);
		execService = Executors.newFixedThreadPool(10);
		
		try {
			dbManager = new BerkeleyDBManager(id);			
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (DatabaseException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
				
		try {
			((SlaveListener)listenerThread).notifyMaster(clientListenerPort);
		} catch (SocketException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		} catch (IOException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}
		listenerThread.start();
		while(true){
			Socket s = connManager.connectTo(null);//waiting for some client
			ClientHandler ch = new ClientHandler(this, s,dbManager);
			//System.out.println("got connection from port " + s.getPort());
			cHandlers.add(ch);
			//Thread t = new Thread(ch);//ch.run();
			execService.execute(ch);
			//t.start();
		}
	}

	@Override
	public void sendData(IMessage message, IMachine machine) {
		byte[] serializedData = message.serialize();		
		connManager.writeData(machine, serializedData);
		connManager.disconnectFrom(machine);
	}
	
	public static void main(String[] args)//arg0 is the ip and arg1 is the identifier, arg2 is the master ip
	{
		Slave s = new Slave(args[0],args[1],args[2]);		
	}
	
	@Override
	public synchronized void notifyTermination(ClientHandler ch) {
		// TODO Auto-generated method stub
		cHandlers.remove(ch);		
		
	}

	@Override
	public void initialize(String ip, String id) {
		// TODO Auto-generated method stub
		
	}

}
